#include <iterator>
#include <unistd.h>
#include "MarkupSTL.h"
#include "global.h"
#include "cmd.h"
#include "Game.h"
#include "RedisClient.h"
#include "Logic.h"
#include "json.h"
#include "globalExtern.h"
#include "Host.h"
#include "util.h"
#include "NodeServer.h"
#include "2850_gamerecord.pb.h"


#define  MOD_UID (100)
char* CGameServer::s_szexchange = "gamerecord_flow";
char* CGameServer::s_szqueue = "gamerecord";

CGameServer::CGameServer():m_pthread(NULL), m_mqchannel(1), m_nodecount(0), m_nodeindex(-1), m_exitflag(false), m_initflag(false), m_cachecheckflag(false)
{
	m_dictareatimezoneinmin["IN"] = 330;
	m_dictareatimezoneinmin["IND"] = 330;
	m_dictareatimezoneinmin["PH"] = 480;
	m_dictareatimezoneinmin["PHL"] = 480;
	m_dictareatimezoneinmin["BR"] = -180;
	m_dictareatimezoneinmin["BRA"] = -180;
	m_dictareatimezoneinmin["VN"] = 420;
	m_dictareatimezoneinmin["VNM"] = 420;
	m_dictareatimezoneinmin["PK"] = 300;
	m_dictareatimezoneinmin["PAK"] = 300;
	m_dictareatimezoneinmin["BD"] = 360;
	m_dictareatimezoneinmin["BGD"] = 360;
	m_dictareatimezoneinmin["BAN"] = 360;
	m_dictareatimezoneinmin["KH"] = 420;
	m_dictareatimezoneinmin["KHM"] = 420;
	m_dictareatimezoneinmin["LA"] = 420;
	m_dictareatimezoneinmin["LAO"] = 420;
	m_dictareatimezoneinmin["LAOS"] = 420;
	m_dayamountfield.push_back("betamount");
	m_dayamountfield.push_back("winamount");
}

CGameServer::~CGameServer()
{
	if (m_pthread)
	{
		m_exitflag = true;
		m_pthread->join();
		delete m_pthread;
		m_pthread = NULL;
	}

	if (m_mqconn)
		releasemq();
}

int CGameServer::OnBYTimeOut(int Timerid)
{
	switch (Timerid)
	{
	case TIMERID_CACHE_CHECK:
		procTimerCacheCheck();
		break;
	
	default:
		break;
	}
	
	return 0;
}

bool CGameServer::InitGame()
{
	char szconfig[] = "./config.xml";
	if (!ReadCommonConf(szconfig))
		return false;

	return initmq();
}

bool CGameServer::initmq(void)
{
	m_mqconn = amqp_new_connection();
	if (!m_mqconn)
	{
		log_error("create mq conn err");
		return false;
	}
	amqp_socket_t* socket = amqp_tcp_socket_new(m_mqconn);
	if (!socket)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("create mq TCP socket err");
		return false;
	}
	int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
	if (status)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq socket err");
		return false;
	}
	amqp_rpc_reply_t x = amqp_login(m_mqconn, m_mqVhost.c_str(), 0, 131072, 30, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPwd.c_str());
	if (AMQP_RESPONSE_NORMAL != x.reply_type)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPwd.c_str());
		return false;
	}
	amqp_channel_open_ok_t* t = amqp_channel_open(m_mqconn, m_mqchannel);
	if (!t)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq channel fail");
		return false;
	}
	// 	if (0 == t->channel_id.len || !t->channel_id.bytes)
	// 	{
	// 		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
	// 		amqp_destroy_connection(m_mqconn);
	// 		m_mqconn = NULL;
	// 		log_error("open mq channel fail");
	// 		return false;
	// 	}
	// 	log_debug("mq channle:%s", std::string((char*)t->channel_id.bytes, t->channel_id.len).c_str());

	amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x1.reply_type)
	{
		releasemq();
		log_error("mq channel amqp_get_rpc_reply fail");
		return false;
	}

	return true;
}

bool CGameServer::mqbind(void)
{
	if (!(m_nodecount > 0))
		return false;
	if (!(m_nodeindex < m_nodecount))
		return false;

	//bool bindflag = false;
	char szqueue[50] = { 0 };
	::snprintf(szqueue, sizeof(szqueue) - 1, "%s%u", s_szqueue, TGlobal::_svid);
	for (int i = 0; i < MOD_UID; ++i)
	{
		if (m_nodeindex == i % m_nodecount)
		{
			amqp_queue_declare_ok_t* r = amqp_queue_declare(m_mqconn, m_mqchannel, amqp_cstring_bytes(szqueue), false, true, false, false, amqp_empty_table);
			if (!r)
			{
				releasemq();
				log_error("mq amqp_queue_declare fail");
				return false;
			}

			amqp_rpc_reply_t x2 = amqp_get_rpc_reply(m_mqconn);
			if (AMQP_RESPONSE_NORMAL != x2.reply_type)
			{
				releasemq();
				log_error("mq amqp_queue_declare fail");
				return false;
			}

			amqp_queue_bind(m_mqconn, m_mqchannel, amqp_cstring_bytes(szqueue), amqp_cstring_bytes(CGameServer::s_szexchange), amqp_cstring_bytes(std::to_string(i).c_str()), amqp_empty_table);
			amqp_rpc_reply_t x = amqp_get_rpc_reply(m_mqconn);
			if (AMQP_RESPONSE_NORMAL != x.reply_type)
			{
				releasemq();
				log_error("mq amqp_queue_bind fail,exchange:%s,queue:%s,routing:%d", CGameServer::s_szexchange, szqueue, i);
				return false;
			}
			//bindflag = true;
			log_debug("amqp_queue_bind ok,exchange:%s,queue:%s,routing:%d", CGameServer::s_szexchange, szqueue, i);

			amqp_basic_consume(m_mqconn, m_mqchannel, amqp_cstring_bytes(szqueue), amqp_empty_bytes, false, true, false, amqp_empty_table);
			amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
			if (AMQP_RESPONSE_NORMAL != x1.reply_type)
			{
				releasemq();
				log_error("mq amqp_queue_bind fail,exchange:%s,queue:%s", CGameServer::s_szexchange, szqueue);
				return false;
			}
		}
	}

// 	if (bindflag)
// 	{
// 		amqp_basic_consume(m_mqconn, m_mqchannel, amqp_cstring_bytes(CGameServer::s_szqueue), amqp_empty_bytes, false, true, false, amqp_empty_table);
// 		amqp_rpc_reply_t x = amqp_get_rpc_reply(m_mqconn);
// 		if (AMQP_RESPONSE_NORMAL != x.reply_type)
// 		{
// 			releasemq();
// 			log_error("mq amqp_queue_bind fail,exchange:%s,queue:%s", CGameServer::s_szexchange, CGameServer::s_szqueue);
// 			return false;
// 		}
// 	}
// 	else
// 	{
// 		log_error("service find no index, srv id,%d,index,%d,service num,%d", TGlobal::_svid, m_nodeindex, m_nodecount);
// 		return false;
// 	}

	return true;
}
void CGameServer::releasemq(void)
{
	if (m_mqconn)
	{
		amqp_channel_close(m_mqconn, m_mqchannel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
	}
	return;
}

void CGameServer::run(void)
{
	struct timeval timeout;
	timeout.tv_sec = 2;
	timeout.tv_usec = 0;
	amqp_frame_t frame;
	while (!m_exitflag)
	{
		if (!m_mqconn)
		{
			if (!initmq())
			{
				log_error("mq reconnect fail,sleep 1");
				::sleep(1);
				continue;
			}
			if (!mqbind())
			{
				log_error("mq rebind fail,sleep 1");
				::sleep(1);
				continue;
			}
		}

		amqp_envelope_t envelope;
		amqp_maybe_release_buffers(m_mqconn);
		amqp_rpc_reply_t ret = amqp_consume_message(m_mqconn, &envelope, &timeout, 0);
		if (AMQP_RESPONSE_NORMAL == ret.reply_type)
		{
			procMessage((const char*)envelope.message.body.bytes, envelope.message.body.len);
			amqp_destroy_envelope(&envelope);
		}
		else
		{
			if (AMQP_RESPONSE_LIBRARY_EXCEPTION == ret.reply_type)
			{
				if (AMQP_STATUS_TIMEOUT == ret.library_error)
				{
				}
				else if (AMQP_STATUS_HEARTBEAT_TIMEOUT == ret.library_error)
				{
					log_debug("heart beat");
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
				else if (AMQP_STATUS_SOCKET_ERROR == ret.library_error)
				{
					log_error("sock err");
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
				else if (AMQP_STATUS_CONNECTION_CLOSED == ret.library_error)
				{
					log_error("connection closed");
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
				else if (AMQP_STATUS_UNEXPECTED_STATE == ret.library_error)
				{
					if (AMQP_STATUS_OK != amqp_simple_wait_frame(m_mqconn, &frame))
					{
						log_error("amqp_simple_wait_frame fail");
						releasemq();
						if (!initmq())
						{
							log_error("mq reconnect fail");
							continue;
						}
						if (!mqbind())
						{
							log_error("mq rebind fail");
							continue;
						}
						continue;
					}

					if (AMQP_FRAME_METHOD == frame.frame_type) 
					{
						switch (frame.payload.method.id) 
						{
						case AMQP_BASIC_ACK_METHOD:
							/* if we've turned publisher confirms on, and we've published a
							 * message here is a message being confirmed.
							 */
							break;
						case AMQP_BASIC_RETURN_METHOD:
							/* if a published message couldn't be routed and the mandatory
							 * flag was set this is what would be returned. The message then
							 * needs to be read.
							 */
						{
							amqp_message_t message;
							ret = amqp_read_message(m_mqconn, frame.channel, &message, 0);
							if (AMQP_RESPONSE_NORMAL != ret.reply_type) 
							{
								break;
							}

							amqp_destroy_message(&message);
						}

						break;

						case AMQP_CHANNEL_CLOSE_METHOD:
							/* a channel.close method happens when a channel exception occurs,
							 * this can happen by publishing to an exchange that doesn't exist
							 * for example.
							 *
							 * In this case you would need to open another channel redeclare
							 * any queues that were declared auto-delete, and restart any
							 * consumers that were attached to the previous channel.
							 */
						{
							log_error("channel close method");
							releasemq();
							if (!initmq())
							{
								log_error("mq reconnect fail");
								continue;
							}
							if (!mqbind())
							{
								log_error("mq rebind fail");
								continue;
							}
						}
						break;

						case AMQP_CONNECTION_CLOSE_METHOD:
							/* a connection.close method happens when a connection exception
							 * occurs, this can happen by trying to use a channel that isn't
							 * open for example.
							 *
							 * In this case the whole connection must be restarted.
							 */
						{
							log_error("connection close method");
							releasemq();
							if (!initmq())
							{
								log_error("mq reconnect fail");
								continue;
							}
							if (!mqbind())
							{
								log_error("mq rebind fail");
								continue;
							}
						}
						break;
						default:
							log_error("An unexpected method was received %u", frame.payload.method.id);
							break;
						}
					}
				}
				else
				{
					log_error("unexpected err,%d", ret.library_error);
				}
			}
			else if (AMQP_RESPONSE_SERVER_EXCEPTION == ret.reply_type)
			{
				if (AMQP_CHANNEL_CLOSE_METHOD == ret.reply.id || AMQP_CONNECTION_CLOSE_METHOD == ret.reply.id)
				{
					log_error("reply.id:%x", ret.reply.id);
					releasemq();
					if (!initmq())
					{
						log_error("mq reconnect fail");
						continue;
					}
					if (!mqbind())
					{
						log_error("mq rebind fail");
						continue;
					}
				}
			}
		}
		if (m_cachecheckflag)
		{
			chachCheck();
			m_cachecheckflag = false;
		}
	}

	return;
}

int CGameServer::procMessage(const char* data, const unsigned int len)
{
	if (!data || 0 == len)
		return 1;

	gamerecord::GameRecordData msg;
	if (!msg.ParseFromArray(data, len))
	{
		log_error("parse pb fail,len:%u", len);
		return 2;
	}
	
	unsigned int uid = msg.uid();
	unsigned int agencyid = msg.agencyid();
	unsigned int game = msg.game();
	int64 bet = msg.bet();
	int64 win = msg.win();
	if (0 == uid || 0 == agencyid || 0 == game)
	{
		log_error("wrong param,uid,%u,agencyid,%u,game,%u", uid, agencyid, game);
		return 3;
	}
	std::string area;
	std::string monthlisttname;
	std::string gamerecordcountname;

	std::map<unsigned int, std::string>::const_iterator itor = m_agencyarea.find(agencyid);
	if (m_agencyarea.end() == itor)
	{
		int nodataflag = 0;
		bool ret = m_dbHandlerAgency.getAgencyArea(agencyid, area, nodataflag);
		if (!ret || nodataflag || area.empty())
		{
			log_error("get area fail,ret,%d,nodataflag,%d,area,%s",(int)ret, nodataflag, area.c_str());
			return 3;
		}

		std::size_t found = area.find(',');
		if (found != std::string::npos)
		{
			log_info("more then one country,%s", area.c_str());
			area = area.substr(0, found);
			log_info("one country,%s", area.c_str());
		}
		std::transform(area.begin(), area.end(), area.begin(), [](unsigned char c) { return std::toupper(c); });

		m_agencyarea.emplace(agencyid, area);
	}
	else
	{
		area = itor->second;
	}
	
	std::map<std::string, int>::const_iterator itor1 =  m_dictareatimezoneinmin.find(area);
	if (m_dictareatimezoneinmin.end() == itor1)
	{
		log_error("get no tiem zone,uid,%u,area,%s",uid, area.c_str());
		return 4;
	}
	
	monthlistTable(game, monthlisttname); 
	if (m_monthlistables.end() == m_monthlistables.find(monthlisttname))
	{
		if (!m_dbHandlerAssistant.createMonthListTable(monthlisttname))
		{
			log_error("create monthlist table fail,%s", monthlisttname.c_str());
			return 5;
		}
		
		m_monthlistables.emplace(monthlisttname);
	}
	
	gamerecordCountTable(game, g_record_count_mod, uid, gamerecordcountname);
	if (m_gamerecordcounttables.end() == m_gamerecordcounttables.find(gamerecordcountname))
	{
		if (!m_dbHandlerAssistant.createGameRecordCountTable(gamerecordcountname))
		{
			log_error("create gamerecord count table fail,%s", gamerecordcountname.c_str());
			return 5;
		}
		
		m_gamerecordcounttables.emplace(gamerecordcountname);
	}
	
	time_t tt = ::time(NULL);
	std::string ym;
	unsigned int mseq = 0;
	std::string recordtname;

	_u64_ rtime = msg.timestamp();
	if (0 == rtime)
		rtime = tt;

	if (0 != caclym(monthlisttname, game, rtime, itor1->second, ym, mseq))
	{
		log_error("caclym error,time,%llu,ym,%s,mseq,%d,tname,%s,game,%u", rtime, ym.c_str(), mseq, monthlisttname.c_str(), game);
		return 6;
	}

    tableName(game, ym, uid, g_uid_mod, recordtname);
	std::set<std::string>::const_iterator itor2 = m_recordtables.find(recordtname);
	if (m_recordtables.end() == itor2)
	{
		if (!m_dbHandler.createRecordTable(recordtname))
		{
			log_error("add table fail,%s", recordtname.c_str());
			return 5;
		}
		m_recordtables.emplace(recordtname);
		if (!m_dbHandlerMaintenance.addTableRecord(recordtname, game, ::atoi(ym.c_str()), uid%g_uid_mod, tt))
		{
			log_error("record table fail,%s", recordtname.c_str());
		}	
	}
 
	bool intable = false;
	unsigned int recordcount = 0;
	std::map<unsigned int, stRecordCache>& recordcache = m_recordcache[game];
	std::map<unsigned int, stRecordCache>::iterator itor3 = recordcache.find(uid);
	if (recordcache.end() != itor3)
	{
		recordcount = itor3->second.m_count;
		intable = true;
	}
	else
	{ 
		stRecordCache cache;

		unsigned int rcount = 0;
		int flag = 0;
		if (m_dbHandlerAssistant.getUserRecordCount(gamerecordcountname, uid, g_record_count_mod, rcount, flag))
		{
			cache.m_count = rcount;
			std::pair<std::map<unsigned int, stRecordCache>::iterator, bool> ret = recordcache.emplace(uid, cache);
			if (ret.second)
			{
				itor3 = ret.first;
			}
			else
			{
				log_error("map insert err,uid,%d", uid);
				return 7;
			}

			recordcount = rcount;
			intable = true;
			if (flag > 0)
			{
				 intable = false;
			}
		}
		else
		{
			log_error("get recordcount fail,%d", uid);
			return 5;
		}
	}
	if (recordcache.end() == itor3)
	{
		log_error("unexcept itor,uid,%u", uid);
		return 8;
	}

    bool newkey = false;
	int dayfirstflag = 0;
	unsigned int ymd = 0;
    std::string dayamountkey;
	timestampToymd(rtime, itor1->second, ymd);
    redisDayAmountKey(uid, game, ymd, dayamountkey);
	if (0 == itor3->second.m_lastday)
	{
		std::vector<std::string> val;
		int nret = m_gamerecordRedis->Hmget(dayamountkey, m_dayamountfield, &val);
		log_debug("hmget,ret,%d,value size,%u",nret, val.size());

		itor3->second.m_dayamountbet = 0;
		itor3->second.m_dayamountwin = 0;
		itor3->second.m_lastday = ymd;
		newkey = true;
		if (0 == nret)
		{
			if (0 == val.size())
			{
				dayfirstflag = 1;
				log_debug("no field value");
			}
			else if (2 == val.size())
			{		
				if (!val[0].empty())
				{
				    itor3->second.m_dayamountbet = ::atoll(val[0].c_str());
				}
				if (!val[1].empty())
				{
					itor3->second.m_dayamountwin = ::atoll(val[1].c_str());
				}

				if (val[0].empty() && val[1].empty())
				{
					dayfirstflag = 1;
					log_info("all field empty");
				}

				log_debug("betamount,%lld, winamount,%lld,", itor3->second.m_dayamountbet, itor3->second.m_dayamountwin);
			}
			else
			{
				log_error("unexcept size,%u", val.size());
			}
		}
		else
		{
			log_error("redis fail,%d", nret);
		}
	}
	else if (itor3->second.m_lastday != ymd)
	{
		itor3->second.m_dayamountbet = 0;
		itor3->second.m_dayamountwin = 0;

		itor3->second.m_lastday = ymd;
		newkey = true;
		dayfirstflag = 1;
	}
	
	itor3->second.m_dayamountbet += bet;
	itor3->second.m_dayamountwin += win;

    ++recordcount;
	log_debug("record,table,%s,uid,%u,useq,%u,roundid,%s,bet,%lld,betact,%d,win,%lld,winact,%d,betamount,%lld, winamount,%lld,tt,%llu,dayfirstflag,%d,detail len,%u,ext len,%u", recordtname.c_str(), uid, recordcount, msg.roundid().c_str()
	, bet, msg.betact(), win, msg.winact(), itor3->second.m_dayamountbet, itor3->second.m_dayamountwin, rtime, dayfirstflag, msg.detail().size(), msg.ext().size());
	if (!m_dbHandler.addRecord(recordtname, uid, recordcount, msg.roundid(), bet, msg.betact(), win, msg.winact() 
		, itor3->second.m_dayamountbet, itor3->second.m_dayamountwin, rtime, dayfirstflag, msg.detail(), msg.ext()))
	{
	    log_debug("record fail,table,%s,uid,%u,useq,%u,roundid,%s,bet,%lld,betact,%d,win,%lld,winact,%d,betamount,%lld, winamount,%lld,tt,%llu", recordtname.c_str(), uid, recordcount, msg.roundid().c_str()
	    , bet, msg.betact(), win, msg.winact(), itor3->second.m_dayamountbet, itor3->second.m_dayamountwin, rtime);
		return 5;
	}

    _u64_ seq = m_dbHandler.getLastSqlId();
	if (intable)
	{
		m_dbHandlerAssistant.addUserRecordCount(gamerecordcountname, uid, g_record_count_mod, 1);
	}
	else
	{
		m_dbHandlerAssistant.insertAndaddUserRecordCount(gamerecordcountname, uid, g_record_count_mod, 1, itor1->second);
	}

	itor3->second.m_lastflowtime = tt;
	++(itor3->second.m_count);

	_u64_ index = 0;
	if (mseq > 0)
	{
		index = (((_u64_)mseq << 48) | seq);
	}

	if (index > 0)
	{
		gamerecord::GameRecordDataCache recordcache;
		recordcache.set_uid(uid);
		recordcache.set_game(game);
		recordcache.set_roundid(msg.roundid());
		recordcache.set_bet(bet);
		recordcache.set_betact(msg.betact());
		recordcache.set_win(win);
		recordcache.set_winact(msg.winact());
		recordcache.set_timestamp(rtime);
		recordcache.set_detail(msg.detail());
		recordcache.set_daybetamount(itor3->second.m_dayamountbet);
		recordcache.set_daywinamount(itor3->second.m_dayamountwin);
		recordcache.set_timezoneinmin(itor1->second);
		recordcache.set_firstofdayflag(dayfirstflag);
		recordcache.set_recordcount(recordcount);
		recordcache.set_cacheindex(index);
        recordcache.set_ext(msg.ext());

		m_pHost->MtSendtoSrv(gamerecord::GAMERECORD_CMD_RECORD_CACHE_PUSH, SERVER_TYPE_GAMERECORD, &recordcache, uid);
	}

    std::vector<std::string> upval;
	upval.emplace_back(std::to_string(itor3->second.m_dayamountbet));
	upval.emplace_back(std::to_string(itor3->second.m_dayamountwin));
	int nret = m_gamerecordRedis->Hmset(dayamountkey, m_dayamountfield, upval);
	log_debug("hmget,ret,%d",nret);
	if (newkey)
	{
		m_gamerecordRedis->Expire(dayamountkey, g_redis_dayamount_dur);
	}
	
	return 0;
}

int CGameServer::ProcEventPublishService(int ev_id, SpiderEvent* pEvent)
{
	if (!(pEvent && pEvent->evParam.ptr))
		return -1;

	if (m_pthread)
		return 0;

	ServiceConf* pserviceConf = (ServiceConf*)pEvent->evParam.ptr;
	if (SERVER_TYPE_GAMERECORD_STORAGE == pserviceConf->server_type)
	{
		ServiceTypeList::const_iterator itor = m_pHost->m_pNode->m_serviceTypeList.find(SERVER_TYPE_GAMERECORD_STORAGE);
		if (m_pHost->m_pNode->m_serviceTypeList.end() != itor && itor->second.size() > 0)
		{
			int index = 0;
			for (ServiceVect::const_iterator itor1 = itor->second.begin(); itor->second.end() != itor1; ++itor1, ++index)
			{
				if (TGlobal::_svid == itor1->svid)
				{
					break;
				}
			}


			if (index < itor->second.size())
			{
				m_nodecount = itor->second.size();
				m_nodeindex = index;

				log_debug("srv id,%d,index,%d,service num,%d", TGlobal::_svid, index, itor->second.size());
				if (m_nodecount > 0)
				{
					if (mqbind())
					{
						m_pthread = new std::thread([this](void) {
							run();
							});
					}
					else
					{
						exit(-1);
						return false;
					}
				}
				else
				{
					log_error("wrong srv num,srv id,%d,index,%d,service num,%d", TGlobal::_svid, index, itor->second.size());
				}
			}
			else
			{
				log_error("service find no srv id,srv id,%d", TGlobal::_svid);
			}
			log_debug("service pulish ready, srv id,%d", TGlobal::_svid);
		}
		else
		{
			log_error("service pulish not ready");
		}
	}
	return 0;
}

int CGameServer::ProcessPacket(SocketHandler * pHandler, NETInputPacket* inputPkg)
{
	unsigned int cmd = inputPkg->GetCmdType();
	switch (cmd)
	{
	default:
		break;
	}

	return 0;
}

bool CGameServer::monthdis(const std::string& m1, const std::string& m2, int& dis)
{
	static unsigned int s_mstrlen = 2;
	if (!(m1.length() >= 6 && m2.length() >= 6))
		return false;
	if (!(::atoi(m1.c_str()) > 0 && ::atoi(m2.c_str()) > 0))
		return false;

	std::string m1y = m1.substr(0, m1.length() - s_mstrlen);
	std::string m1m = m1.substr(m1.length() - s_mstrlen, string::npos);
	std::string m2y = m2.substr(0, m2.length() - s_mstrlen);
	std::string m2m = m2.substr(m2.length() - s_mstrlen, string::npos);
	int im1y = ::atoi(m1y.c_str());
	int im1m = ::atoi(m1m.c_str());
	int im2y = ::atoi(m2y.c_str());
	int im2m = ::atoi(m2m.c_str());
	if (!(im1y > 0 && im2y > 0))
		return false;
	if (!(im1m >= 1 && im1m <= 12 && im2m >= 1 && im2m <= 12))
		return false;

	dis = (im2y - im1y) * 12 + (im2m - im1m);
	return true;
}

int CGameServer::caclym(const std::string& tname, const unsigned int game, time_t tt, const int timezoneinmin, std::string& ym, unsigned int& mseq) 
{
// 	time_t tt = ::time(NULL);
// 	std::string ym;
// 	unsigned int mseq = 0;

	yearMonthStr(tt, timezoneinmin, ym);
	std::map<std::string, unsigned int>& mindex = m_gamemindex[game];
	std::map<unsigned int, std::string>& indexm = m_gameindexm[game];
	std::map<std::string, unsigned int>::const_iterator itor = mindex.find(ym);
	if (mindex.end() == itor)
	{
		int nodataflag = 0;
		if (m_dbHandlerAssistant.getYmInfo(tname, ym, mseq, nodataflag))
		{
			mindex.emplace(ym, mseq);
			indexm[mseq] = ym;
		}
		else
		{
			if (1 == nodataflag)
			{
				std::pair<std::string, unsigned int>& firstm = m_gamefirstm[game];
				if (firstm.first.empty() || firstm.second == 0)
				{
					int nodataflag1 = 0;
					if (!m_dbHandlerAssistant.getFirstMonth(tname, firstm, nodataflag1))
					{
						if (1 == nodataflag1)
						{
							time_t ttcur = ::time(NULL);
							if (tt <= ttcur)
								firstm.first = ym;
							else
							{
								std::string ymcur;
								yearMonthStr(ttcur, timezoneinmin, ymcur);
								firstm.first = ymcur;
							}
							firstm.second = 1;
						}
						else
						{
							log_error("get first month fail,%s", ym.c_str());
							return 1;
						}
					}
				}

				if (ym == firstm.first)
				{
					mseq = firstm.second;
				}
				else
				{
					int dis = 0;
					bool bcal = monthdis(firstm.first, ym, dis);
					if (!bcal || dis < 0)
					{
						log_error("monthdis fail,%s, first:%s,dis:%d", ym.c_str(), firstm.first.c_str(), dis);
						return 2;
					}
					mseq = firstm.second + dis;
				}

				if (0 == mseq)
				{
					log_error("wrong mseq,%s, first:%s,seq:%u", ym.c_str(), firstm.first.c_str(), mseq);
					return 2;
				}

				if (m_dbHandlerAssistant.addYearMonth(tname, ym, mseq))
				{
					mindex.emplace(ym, mseq);
			        indexm[mseq] = ym;
				}
				else
				{
					log_error("add year-month fail,%s", ym.c_str());
					return 1;
				}
			}
			else
			{
				log_error("get year-month fail,%s", ym.c_str());
				return 1;
			}
		}
	}
	else
	{
		mseq = itor->second;
	}

	return 0;
}

bool CGameServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}


	if (!markup.FindElem("mq"))
	{
		log_error("Not Found Elem [mq] in xml[%s]", strxml);
		return false;
	}
	else
	{
		m_mqIp = markup.GetAttrib("ip");
		m_mqPort = atoi(markup.GetAttrib("port").c_str());
		m_mqVhost = markup.GetAttrib("vhost");
		m_mqUsername = markup.GetAttrib("username");
		m_mqPwd = markup.GetAttrib("password");
	}

	if (m_mqIp.empty() || m_mqVhost.empty() || m_mqUsername.empty() || m_mqPwd.empty() || 0 == m_mqPort)
		return false;

	log_debug("ReadCommonConf succ!!");

	return true;
}

int CGameServer::ProcEventGameInit(int ev_id, SpiderEvent* pEvent)
{
	if (!m_initflag)
	{
		if (0 != m_pHost->InitMtSend())
		{
			log_error("InitMtSend fail");
			exit(-1);
			return -1;
		}
		
		StartTimer(TIMERID_CACHE_CHECK, g_cache_check_dur, true);
		m_initflag = true;
	}

	return 0;
}

int CGameServer::procTimerCacheCheck(void)
{
	m_cachecheckflag = true;

	return 0;
}

void CGameServer::chachCheck(void)
{
	unsigned int tc = 0;
	unsigned int pc = 0;
	time_t tt = ::time(NULL);

	for (std::map<unsigned int, std::map<unsigned int, stRecordCache> >::iterator itor =  m_recordcache.begin(); m_recordcache.end() != itor; ++itor)
	{
		for (std::map<unsigned int, stRecordCache>::iterator itor1 = itor->second.begin(); itor->second.end() != itor1;)
	    {
			if (tt - itor1->second.m_lastflowtime > g_cache_dur)
		    {
				itor->second.erase(itor1++);
				++pc; 
		    }
		    else
		    {
			    ++itor1;
		    }
			++tc;
	    }
	}
	
	log_debug("check proc,tc,%u,pc,%u,t,%u", tc, pc, (::time(NULL) - tt));
}

void CGameServer::monthlistTable(const unsigned int game, std::string& tablename)
{
	char name[64] = { 0 };
	::snprintf(name, sizeof(name)-1, "month_list_%u", game);
	tablename = name;

	return;
}

void CGameServer::gamerecordCountTable(const unsigned int game, const unsigned int mod, const unsigned int uid, std::string& tablename)
{
	char name[64] = { 0 };
	::snprintf(name, sizeof(name)-1, "user_gamerecord_count_%u_%u", game, (uid%mod));
	tablename = name;

	return;
}

void CGameServer::redisDayAmountKey(const unsigned int uid, const unsigned int game, const unsigned int ymd, std::string& rkey)
{
	char name[64] = { 0 };
	::snprintf(name, sizeof(name)-1, "dayamount:%u:%u:%u", game, ymd, uid);
	rkey = name;

	return;
}